AWS Summit 2014 Tokyo「Amazon Kinesis Deep Dive」レポート

#AWS

虎塚

2014.08.27

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、虎塚です。

先月のAWSサミットのセッション「Amazon Kinesis Deep Dive」をタイムシフト視聴しましたので、レポートします。

講師は、大谷晋平さん（アマゾンデータサービスジャパン）と堀剛さんです。

はじめに

AWSサミット東京の初日、Kinesisの東京リージョンローンチが発表された。

これに合わせて、Kinesisを実際に開発しているDevelopment Managerの堀さんを招待した。

Kinesisの概要や事例を大谷さんから紹介した後、Kinesisにかける思いや、実際にどういったところで使えるのかを堀さんから聴く。

Kinesisの新規ローンチ

3つの大きな発表。

東京リージョンでKinesisローンチ: これまでus-east-1かus-east-2だけだった
Fluentdプラグイン: データ入出力のプラグインとして、日本のOSSで重要な位置づけを占めるログコレクタのプラグインを開発。既存のデータをKinesisにシームレスに流し込める
MQTTアダプター: AWS顧客の要望を受けて、軽量なデータ転送フォーマット・MQTTに対応。OSSのライブラリを開発。既存システムや新規システムとKinesisとの連携がしやすくなった。

事例紹介

ガリバーさま: Drive+

LINEと連携するサービスをローンチ予定。車を駐車した場所を忘れてしまう問題に対して、LINEにスタンプを送ると地図の場所が返ってくる画期的なサービスを提供。

RedshiftやDynamoDBなど、AWSサービスをふんだんに使っている。キーポイントが、Kinesisよるリアルタイムなデータ連携。これによって、他サービスと連携したデータを短期間で作り出せる。

Pencilさま: リアルタイムユーザーモニター

開発が完了し、これから本番稼働。エンドユーザがWebサイトをマウスでどのようにクリックで辿ったかなどを、リアルタイムにモニタし、画面遷移やマウスの軌跡を追跡、取得する。

構想は元々あったが、無理だといわれていた。それがKinesisによって実現できた。開発費も4分の1にできた。

SmartInsightさま: Beacon

会場にBeaconを設置し、自分の名前を登録しておくと、自分とBeaconを距離を計測し、pushして送るサービス。

AWSサミット会場のブースでも、実験的に25台設置している。ユーザが会場を歩き回ると、アプリ上で集客や動線をヒートマップのようなUIで見ることができる。

単純な動向の可視化から、ロケーションを見た上でサービス提供するための基盤を開発されている。データをKinesisで受け取るところがキーとなる。集めたデータはリコメンデーションエンジン（EMR）を通して、最終的にコンテキストを踏まえたリコメンデーションをpushする。

従来はバッチで実現していたが、最後のピースとしてKinesisを使うことで、より間隔の短い鮮度の高いデータを扱えるようになった。その結果、精度の高いユーザ解析ができるようになった。

SUPERCELLさま（US）

ゲームエンジンサーバから送信されるすべてのエンドユーザの挙動（タップやスライド）をKinesisに送信し、大規模な分析を実施。リアルタイムにダッシュボードをモニタしている。

bizoさま（US）

既存システムではバッチで行っていた処理であるデータパイプラインとレポートのインフラを、Kinesisへ置き換えた。鮮度の高いデータを収集し、システム運用リソースを削減した。また、解析頻度を上げることで効果の高い分析を行っている。

Amazon Kinesisについて

ここからは、Amazon Kinesis開発マネージャーである堀さんのお話。堀さんの前職は測定サービス開発マネージャー。

なぜリアルタイム？

堀さんは前職で測定（Metering）サービスの仕事をしていた。AWSは、使った分だけ課金されるサービスのため、オペレーションを測定する必要があった。

AWSのデータ量は、毎秒数千万レコード、毎時数テラバイト。月末には大量のデータ処理を、100%の正確性で実施する必要がある。データウェアハウスの観点から説明すると、毎時数百万のファイルからデータが入り、毎日100以上のバッチが動き、毎日数百ユーザから数百クエリが来るようなシステム。

AWS 測定サービス: DynamoやS3のサーバからレコードが入ってきて、データをS3に入れる; Hadoopでデータをアグリゲートして、次のサービスへ送る

測定サービスでの課題: スケールの課題: AWSのサービスはどんどん増えているので、スケールが必要; リアルタイム性の課題: 1時間より10分、10分より1分の単位で計測したい; 運用コストの課題: 大きくなればなるほど大変

要求の変化: 従来は、ただ毎時毎日の大量データを処理したかった; 今は、リアルタイムに早く意思決定するために、あらゆるデータを片っ端から（keep everything）、拡張性のあるシステムで、複製の目的に応じて1つのデータをさまざまなサービスで並行処理したい

Amazon Kinesis概要

Amazon Kinesisとは: 大規模なストリーミングデータをリアルタイムで収集、処理する; 1時間あたり数十万のソースから数百TBのデータを収集する; データは複数Availability Zoneに保存されるので、高い信頼性と耐久性を持つ

Kinesis構成内容: 端末（サーバ、モバイル等）からデータをStreamに入れると、複数のAvailability Zoneに24時間保存される; そのデータをS3にアーカイブしたり、RedshiftでBIしたり、機械学習に使ったり、次のKinesisに繋いだりできる; Streamは1つ以上のShardで構成される: Shardは入力側: 秒時1MB, 1000TPS, データ処理側: 秒間2MB, 5TPSのキャパシティを持つ; Shardを増減させることでスケールを制御できる; 入力データは、複数のAvailability Zoneに24時間保管される

データ入力

様々な方法でデータをKinesisへ入れることができる。

(HTTP) POST
AWS SDK
Flume
Log4J

データ入力方法は、次のとおり。

データを発信するプロデューサーは、PutRecord APIでストリームにデータを入力する
PutRecordでは、データ(base64-encoded-data、StreamName, PartitionKey)を指定する
パーティションキーをもとにshardにデータを分配する
PutRecordが成功した時のAPIの返却値: シーケンス番号、shard番号
データサイズは最大50kB

PutRecord APIは、AWS CLIから使うのが一番簡単な方法なので、試してみてほしい。

Shardとは何か: Streamは複数のShardで構成される; Shardは担当するレンジを持つ; 指定したパーティションキーをMD5でハッシュ化し、その値によってデータがShardに分配される

重要なこととして、分配が偏らないようにパーティションキーを選択する必要がある。仮に、すべてのデータに同じパーティションキーを使うと、shardがいくつあっても1つのshardしか使われないことになる。

パーティションキーについてのTips

データはパーティションキーでshardに分配される。また、Shardにはキャパシティがある

そのため、パーティションキーの数がshardの数よりずっと多いことが望ましい

GUIDやランダムな値を使うといろいろなshardにきれいに分配される
カーディナリティの高いキーを使うべし

シーケンス番号: KinesisがStream内でデータにユニークなシーケンス番号を振る; データもシーケンス番号も不変; シーケンス番号でデータが(24時間以内)何度でも取得できる; 何度データを取得しても、シーケンス番号の順番は不変

データの取得と処理

入力と同様に、Kinesisからデータを様々な方法で取得できる。

HTTP GET
Kinesis Client Library
Kinesis Connector Library
Storm（リアルタイムな分析プラットフォーム）
Amazon EMR

AWS CLIによるデータの取得方法: get-shard-iterator: イテレータを取得; get-records: データを取ると、データとともに次のイテレータを取得できる

どうやって信頼性と拡張性のあるアプリケーションを作るか

Kinesis Client Library (KCL): KCLがshardと同じ数のworker（プロセス）を立ち上げる; workerが均等にEC2上で走るようにロードバランシングする; workerがクラッシュするとKCLが新しいworkerを立ち上げる; shardの増減にあわせてworkerを増減させる; CPUをモニタリングしてAutoScalingを使う; どこまで処理したかを記憶できるチェックポインティングの取得（障害時のリトライなどのため）

これらの面倒な処理をKCLが担ってくれるので、開発者はエンドユーザのためのビジネスロジックに集中できる。

使い方: インタフェースIRecordProcessorについてロジックを実装する。

重複データについてのTips: ネットワークエラーや500エラーが出ると、プロデューサーは最後のチェックポイントから処理をリトライする; ポイント: ユースケースを理解して、冪等なアプリを作る（たとえば課金集計）or 重複を許容する（たとえば統計）を選択すること

冪等なアプリを作るのには時間がかかるので、重複を許容する簡単なアプリケーションを先に作った方が良い場合もある。お客様に早く利用を開始していただくことで、必要なフィードバックをいただける可能性がある。たとえば、冪等よりももっと必要な機能に気づくかもしれない。

Kinesis Connector Library: データの移動を簡単にするためのライブラリ; 4つのインタフェースを実装して使う

アプリの拡張性についてのTips

状況:

1つのデータをいろいろなシステムで使いたい
プロデューサの負荷を減らしたい
データの一貫性を保ちたい（データを2回書き込むケースなどで問題になる）

解決策:

プロデューサーからKinesisにすべてのデータを1回だけ入力する
必要に応じて新しいアプリを足していく
古いアプリは触らない、ダメだったら（足したアプリを）すぐやめるという使い方をする

エラスティックな拡張性

Streamがキャパシティの単位となる。Shardを足したり引いたりすることで、キャパシティを調整する。

SplitShard API: 1つのshardを2つに分けるAPI; 次のshardをどこから始めるかというスタートのハッシュキーを指定する

キャパシティについてのTips: shardにもEC2にもキャパシティがあるので、両方モニタリングすること(AutoScalingが有用）

Kinesisコスト: shard1つで1ヶ月$14, PUTは100万PUTで$0.043, GETは無料

デモンストレーションの紹介: リアルタイムのダッシュボード（※セッションの動画をぜひご覧ください）: HTTPのシミュレーション。WebサイトのリファーをDynamoDBに入れて、リファー数を見るアプリ。; describeするとshardが2つ。shardを1つsplitしてから、もう一度describeすると、shardが3つになる。; AutoScalingを使ってKCLが自動的にworkerを立ち上げる。; キャパシティに余裕が出る。キャパシティが増えると、入ってくるデータの数も増える。; 別のアプリから同じデータをリアルタイムに使って見ることができる。プロデューサーは1回しかデータを入れていないことが重要。